package com.pingan.haofang.searchcloud.index.service.impl;

import com.google.common.collect.Lists;
import com.pingan.haofang.framework.common.functional.lang.control.base.Enums;
import com.pingan.haofang.framework.common.functional.lang.control.bean.BeanCopy;
import com.pingan.haofang.searchcloud.api.StorageCluster;
import com.pingan.haofang.searchcloud.api.constants.EngineType;
import com.pingan.haofang.searchcloud.api.facade.CollectionConfigFacade;
import com.pingan.haofang.searchcloud.api.matedata.FieldMeta;
import com.pingan.haofang.searchcloud.api.matedata.IndexMeta;
import com.pingan.haofang.searchcloud.api.matedata.IndexMetadataHolderFactory;
import com.pingan.haofang.searchcloud.api.matedata.constant.MetadataFieldType;
import com.pingan.haofang.searchcloud.api.schema.IndexSchema;
import com.pingan.haofang.searchcloud.api.schema.SchemaCopyField;
import com.pingan.haofang.searchcloud.api.schema.SchemaField;
import com.pingan.haofang.searchcloud.collection.dao.CollectionDao;
import com.pingan.haofang.searchcloud.collection.domain.Collection;
import com.pingan.haofang.searchcloud.collection.service.CollectionService;
import com.pingan.haofang.searchcloud.common.configuration.ClusterProperties;
import com.pingan.haofang.searchcloud.common.constants.Status;
import com.pingan.haofang.searchcloud.common.dto.OperationResult;
import com.pingan.haofang.searchcloud.common.dto.PageDTO;
import com.pingan.haofang.searchcloud.common.exception.FieldException;
import com.pingan.haofang.searchcloud.common.exception.FieldExceptionItem;
import com.pingan.haofang.searchcloud.common.exception.NotFoundException;
import com.pingan.haofang.searchcloud.common.support.PageSupport;
import com.pingan.haofang.searchcloud.index.constants.IndexConstants;
import com.pingan.haofang.searchcloud.index.dao.IndexDao;
import com.pingan.haofang.searchcloud.index.dao.IndexSpec;
import com.pingan.haofang.searchcloud.index.domain.Index;
import com.pingan.haofang.searchcloud.index.dto.IndexDTO;
import com.pingan.haofang.searchcloud.index.dto.IndexInfoDTO;
import com.pingan.haofang.searchcloud.index.dto.IndexInfoQueryResult;
import com.pingan.haofang.searchcloud.index.dto.IndexQueryDTO;
import com.pingan.haofang.searchcloud.index.dto.IndexQueryResult;
import com.pingan.haofang.searchcloud.index.service.ClusterService;
import com.pingan.haofang.searchcloud.index.service.IndexService;
import com.pingan.haofang.searchcloud.metadata.dao.MetadataDao;
import com.pingan.haofang.searchcloud.metadata.dao.MetadataFieldDao;
import com.pingan.haofang.searchcloud.metadata.domain.Metadata;
import com.pingan.haofang.searchcloud.metadata.domain.MetadataField;
import com.pingan.haofang.searchcloud.metadata.dto.MetadataFieldDTO;
import com.pingan.haofang.searchcloud.metadata.dto.MetadataFieldQueryResult;
import com.pingan.haofang.searchcloud.quartz.dto.TaskQueryDto;
import com.pingan.haofang.searchcloud.user.dao.SearchProjectDao;
import com.pingan.haofang.searchcloud.user.domain.SearchProject;
import com.pingan.haofang.searchcloud.user.dto.SearchUserDto;
import com.pingan.haofang.searchcloud.user.service.SearchUserService;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;


/**
 * 索引Service实现
 *
 * @author baojie371
 * @date 17/5/3
 * @since 1.0.0
 */
@Service
@ManagedResource
public class IndexServiceImpl implements IndexService {

    @Autowired
    private SearchProjectDao searchProjectDao;

    @Autowired
    private IndexDao indexDao;

    @Autowired
    private MetadataDao metadataDao;

    @Autowired
    private SearchUserService searchUserService;
    
    @Autowired
    private MetadataFieldDao metadataFieldDao;

    @Autowired
    private CollectionDao collectionDao;

    @Autowired
    private CollectionConfigFacade collectionConfigFacade;

    @Autowired
    private CollectionService collectionService;

    @Autowired
    private ClusterService clusterService;

    @Override
    public PageDTO<IndexQueryResult> query(IndexQueryDTO query) {

        // 获取索引列表数据
        Page<Index> indexPage = indexDao.findByProjectIdAndStatus(query.getProjectId(), Status.NORMAL.getValue(),
                PageSupport.pageQuery2PageRequest(query));
        PageDTO<IndexQueryResult> result =
                PageDTO.of(indexPage, BeanCopy.toFunction(IndexQueryResult.class, BeanUtils::copyProperties));

        if (CollectionUtils.isNotEmpty(indexPage.getContent())) {

            // 获取Meta信息
            for (IndexQueryResult indexQueryResult : result.getDatas()) {
                // 获取Meta信息
                Metadata metadata = findMetadataByIndexIdAndLastVersion(indexQueryResult.getId());
                fillIndexMetadata(indexQueryResult, metadata);
            }

            // 获取用户名
            List<Long> userIds = indexPage.getContent().stream()
                    .map(Index::getCreateBy).distinct()
                    .collect(Collectors.toList());
            List<SearchUserDto> searchUsers = searchUserService.getUserByIds(userIds);
            
            if (CollectionUtils.isNotEmpty(searchUsers)) {
                Map<Long, String> userNameMap = searchUsers.stream()
                        .collect(Collectors.toMap(SearchUserDto::getId, SearchUserDto::getUserName));
                result.getDatas().forEach(
                        data -> data.setCreateByName(userNameMap.getOrDefault(data.getCreateBy(), "")));
            }
        }

        return result;
    }

    @Override
    public List<IndexDTO> queryByProjectId(long projectId) {
        List<Index> indexPage = indexDao.findByProjectIdAndStatus(projectId, Status.NORMAL.getValue());
        List<IndexDTO> result = new ArrayList<>();
        for (Index index : indexPage) {
            IndexDTO dto = new IndexDTO();
            BeanUtils.copyProperties(index, dto);
            result.add(dto);
        }
        return result;
    }
    
    @Override
    public IndexInfoQueryResult queryById(long id) {

        // 获取索引
        Index index = Optional.ofNullable(indexDao.findOne(id))
                .orElseThrow(() -> new NotFoundException("index not found"));

        // 获取用户信息
        SearchUserDto searchUser = searchUserService.getUserById(index.getCreateBy());
        // 获取Meta信息
        Metadata metaData = findMetadataByIndexIdAndLastVersion(index.getId());

        // 获取fields信息
        List<MetadataField> fields = Optional.ofNullable(metadataFieldDao.findByMetaId(metaData.getId()))
                .orElse(Collections.emptyList());

        // 封装DTO
        return createIndexQueryResult(index, metaData, fields, searchUser);
    }

    @Override
    public IndexQueryResult queryBasicInfoById(long id) {
        // 获取索引
        Index index = Optional.ofNullable(indexDao.findOne(id))
                .orElseThrow(() -> new NotFoundException("index not found"));

        // 获取用户信息
        SearchUserDto searchUser = searchUserService.getUserById(index.getCreateBy());

        // 获取Meta信息
        Metadata metadata = findMetadataByIndexIdAndLastVersion(index.getId());

        IndexQueryResult basicInfo = BeanCopy.of(index, new IndexQueryResult())
                .copy(BeanUtils::copyProperties)
                .copy((s, t) -> t.setShardCount(metadata.getShardCount()))
                .copy((s, t) -> t.setReplicationFactor(metadata.getReplicationFactor()))
                .copy((s, t) -> t.setMaxShardsPerNode(metadata.getMaxShardsPerNode()))
                .copy((s, t) -> t.setHost(metadata.getHost()))
                .copy((s, t) -> t.setConfigVersion(metadata.getConfigVersion()))
                .get();
        basicInfo.setCreateByName(Optional.ofNullable(searchUser)
                .map(SearchUserDto::getUserName)
                .orElse(""));
        return basicInfo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public long save(IndexInfoDTO indexInfo) {

        Objects.requireNonNull(indexInfo, "indexInfo is null");
        Objects.requireNonNull(indexInfo.getBasicInfo(), "basicInfo is null");

        IndexDTO basicInfo = indexInfo.getBasicInfo();
        List<MetadataFieldDTO> fields = indexInfo.getFields();

        // 判断projectId是否存在
        SearchProject searchProject = Optional.ofNullable(searchProjectDao.findOne(basicInfo.getProjectId()))
                .orElseThrow(() -> new NotFoundException("Project not found"));
        if (!searchProject.getStatus().equals(Status.NORMAL.getValue())) {
            throw new NotFoundException("Project not found");
        }

        // 有索引ID则更新，否则新增
        if (basicInfo.getId() == null) {
            return add(basicInfo, fields);
        } else {
            return update(basicInfo, fields);
        }
    }

    @Override
    @Transactional
    @ManagedOperation
    public OperationResult enableIndex(long id, boolean force) {
        Index index = Optional.ofNullable(indexDao.findOne(id))
                .orElseThrow(() -> new NotFoundException("Index Not Found"));
        if (!force && !index.getStatus().equals(Status.NORMAL.getValue())) {
            return OperationResult.of(false, "Index status unnormal");
        }

        // 获取Meta信息
        Metadata metaData = findMetadataByIndexIdAndLastVersion(index.getId());
        if (!force && !metaData.getStatus().equals(Status.STOP.getValue())) {
            return OperationResult.of(false, "Metadata status error");
        }

        // 获取fields信息
        List<MetadataField> fields = Optional.ofNullable(metadataFieldDao.findByMetaId(metaData.getId()))
                .orElse(Collections.emptyList());

        // 创建IndexSchema
        IndexSchema indexSchema = createIndexSchema(index, metaData, fields);
        if (!collectionConfigFacade.create(new StorageCluster(metaData.getHost(), index.getEngineType()),
                indexSchema, formatConfigNameInZk(metaData))) {
            return OperationResult.of(false, "Create Index Schema Failed");
        }

        // 更新Metadata的状态
        List<Metadata> metas = metadataDao.findByStatusAndIndexIdIn(Status.NORMAL.getValue(),
                Lists.newArrayList(index.getId()));

        if (CollectionUtils.isNotEmpty(metas)) {
            Metadata oldMetadata = metas.get(0);
            oldMetadata.setStatus(Status.STOP.getValue());
            metadataDao.saveAndFlush(oldMetadata);

            metaData.setStatus(Status.NORMAL.getValue());
            metadataDao.saveAndFlush(metaData);
        } else {
            // 创建Collection
            String collectionName = formatCollectionName(metaData);
            collectionService.getOrCreateCollection(index, metaData, collectionName);

            metaData.setStatus(Status.NORMAL.getValue());
            metadataDao.saveAndFlush(metaData);

            // 注册Collection
            collectionService.registerCollection(index, collectionName);

        }
        //更新meta信息到Zk
        IndexMeta indexMeta = indexInfo2IndexMeta(index, metaData, fields);
        IndexMetadataHolderFactory.get().register(indexMeta);

        return OperationResult.of(true, "");
    }

    @Override
    public List<IndexQueryResult> queryAllIndexOfStorage(TaskQueryDto dto) {

        List<Index> indexes = indexDao.findAll(IndexSpec.listQuerySpec(dto));

        if (CollectionUtils.isEmpty(indexes)) {
            return Collections.emptyList();
        }

        List<IndexQueryResult> result = indexes.stream()
                .map(data -> BeanCopy.of(data, new IndexQueryResult())
                        .copy(BeanUtils::copyProperties)
                        .set(res -> res.setMetadataStatus(Status.STOP.getValue()))
                        .get())
                .collect(Collectors.toList());

        List<Long> ids = result.stream()
                .map(IndexQueryResult::getId)
                .collect(Collectors.toList());

        List<Metadata> metadatas = metadataDao.findByStatusAndIndexIdIn(Status.NORMAL.getValue(), ids);
        if (CollectionUtils.isNotEmpty(metadatas)) {
            Map<Long, Metadata> metadataMap = metadatas.stream()
                    .collect(Collectors.toMap(Metadata::getIndexId, Function.identity(), (v1, v2) -> v1));
            result.forEach(data -> Optional.ofNullable(metadataMap.get(data.getId()))
                    .ifPresent(metadata -> fillIndexMetadata(data, metadata)));
        }

        // 获取用户名
        List<Long> userIds = result.stream()
                .map(IndexQueryResult::getCreateBy).distinct()
                .collect(Collectors.toList());
        List<SearchUserDto> searchUsers = searchUserService.getUserByIds(userIds);
        if (CollectionUtils.isNotEmpty(searchUsers)) {
            Map<Long, String> userNameMap = searchUsers.stream()
                    .collect(Collectors.toMap(SearchUserDto::getId, SearchUserDto::getUserName));
            result.forEach(data -> data.setCreateByName(userNameMap.getOrDefault(data.getCreateBy(), "")));
        }

        // 将获取不到meta的结果过滤
        return result.stream()
                .filter(data -> data.getMetadataStatus() == Status.NORMAL.getValue())
                .collect(Collectors.toList());

    }

    @Override
    public IndexInfoQueryResult queryByIndexNameOfStorage(String indexName) {

        // 获取索引
        List<Index> indexes = indexDao.findByIndexNameAndStatus(indexName, Status.NORMAL.getValue());
        if (CollectionUtils.isEmpty(indexes)) {
            throw new NotFoundException(String.format("Index %s not found", indexName));
        }
        Index index = indexes.get(0);

        // 获取用户信息
        SearchUserDto searchUser = searchUserService.getUserById(index.getCreateBy());
        // 获取Meta信息
        Metadata metaData = findMetadataByIndexId(index.getId());

        // 获取fields信息
        List<MetadataField> fields = Optional.ofNullable(metadataFieldDao.findByMetaId(metaData.getId()))
                .orElse(Collections.emptyList());

        // 封装DTO
        return createIndexQueryResult(index, metaData, fields, searchUser);
    }

    @Override
    public IndexInfoQueryResult queryByIndexNameAndCollectionNameOfStorage(String indexName, String collectionName) {
        // 获取索引
        List<Index> indexes = indexDao.findByIndexNameAndStatus(indexName, Status.NORMAL.getValue());
        if (CollectionUtils.isEmpty(indexes)) {
            throw new NotFoundException(String.format("Index %s not found", indexName));
        }
        Index index = indexes.get(0);

        // 获取用户信息
        SearchUserDto searchUser = searchUserService.getUserById(index.getCreateBy());

        Metadata metaData = findMetadataByIndexIdAndCollectionName(index.getId(), collectionName);

        // 获取fields信息
        List<MetadataField> fields = Optional.ofNullable(metadataFieldDao.findByMetaId(metaData.getId()))
                .orElse(Collections.emptyList());

        // 封装DTO
        return createIndexQueryResult(index, metaData, fields, searchUser);
    }

    @Override
    public Optional<String> findPKFieldNameOfStorage(String indexName) {

        List<Index> indexes = indexDao.findByIndexNameAndStatus(indexName, Status.NORMAL.getValue());
        if (CollectionUtils.isEmpty(indexes)) {
            return Optional.empty();
        }
        Index index = indexes.get(0);
        // 获取Meta信息
        List<Collection> collections = collectionDao.findByIndexIdInAndStatus(Lists.newArrayList(index.getId()),
                Status.NORMAL.getValue());
        if (CollectionUtils.isEmpty(collections)) {
            return Optional.empty();
        }

        Collection collection = collections.get(0);
        long metaId = collection.getMetaId();

        Metadata metadata = metadataDao.findOne(metaId);
        if (metadata == null) {
            return Optional.empty();
        }
        List<MetadataField> fields = metadataFieldDao.findByMetaId(metadata.getId());
        return fields.stream()
                .filter(MetadataField::getUniqueKey)
                .map(MetadataField::getFieldName)
                .findAny();
    }

    @Override
    @Transactional
    public String getOrCreateCollectionOfStorage(String projectCode, String indexName) {

        // 获取索引
        List<Index> indexes = indexDao.findByIndexNameAndStatus(indexName, Status.NORMAL.getValue());
        if (CollectionUtils.isEmpty(indexes)) {
            throw new NotFoundException(String.format("Index %s not found", indexName));
        }
        Index index = indexes.get(0);

        // 获取Meta信息
        Metadata metaData = findMetadataByIndexIdAndStatusNormal(index.getId());
        String collectionName = formatCollectionName(metaData);

        return collectionService.getOrCreateCollection(index, metaData, collectionName).getCollectionName();
    }

    @Override
    public String getCollectionOfStorage(String projectCode, String indexName) {
        // 获取索引
        List<Index> indexes = indexDao.findByIndexNameAndStatus(indexName, Status.NORMAL.getValue());
        if (CollectionUtils.isEmpty(indexes)) {
            throw new NotFoundException(String.format("Index %s not found", indexName));
        }
        Index index = indexes.get(0);

        return collectionService.getCollection(index)
                .map(Collection::getCollectionName)
                .orElseThrow(() -> new NotFoundException(
                        String.format("Collection Not Found, Index name is: %s", indexName)));
    }

    @Override
    @Transactional
    @ManagedOperation
    public void registerCollectionOfStorage(String projectCode, String indexName, String collectionName) {
        // 获取索引
        List<Index> indexes = indexDao.findByIndexNameAndStatus(indexName, Status.NORMAL.getValue());
        if (CollectionUtils.isEmpty(indexes)) {
            throw new NotFoundException(String.format("Index %s not found", indexName));
        }
        Index index = indexes.get(0);

        List<Collection> collections = collectionDao.findByCollectionName(collectionName);
        if (CollectionUtils.isEmpty(collections)) {
            throw new NotFoundException(String.format("Collection %s not found", collectionName));
        }

        Collection collection = collections.get(0);

        Metadata metadata = metadataDao.findOne(collection.getMetaId());
        if (metadata == null) {
            throw new NotFoundException(String.format("Metadata %d not found", metadata));
        }

        List<MetadataField> metadataFields = metadataFieldDao.findByMetaId(metadata.getId());
        IndexMeta indexMeta = indexInfo2IndexMeta(index, metadata, metadataFields);

        collectionService.registerCollection(index, collectionName);

        IndexMetadataHolderFactory.get().register(indexMeta);
    }

    @Override
    public int countIndex() {
        List<Index> indexList = indexDao.findByStatus(Status.NORMAL.getValue());
        if (CollectionUtils.isNotEmpty(indexList)) {
            return indexList.size();
        }
        return 0;
    }

    @Override
    public List<Index> getByindexIds(List<Long> indexIds) {
        return indexDao.findByIdInAndStatus(indexIds,Status.NORMAL.getValue());
    }
    
    /**
     * 新增
     *
     * @param basicInfo
     * @param fields
     * @return
     */
    private long add(IndexDTO basicInfo, List<MetadataFieldDTO> fields) {

        // 索引名称不允许重复
        List<Index> indexes = indexDao.findByIndexName(basicInfo.getIndexName());
        if (CollectionUtils.isNotEmpty(indexes)) {
            throw new FieldException(new FieldExceptionItem("basicInfo.indexName",
                    "index.indexName.exist"));
        }

        // 保存Index
        Index index = createIndex(basicInfo);
        index = indexDao.saveAndFlush(index);

        // 保存Metadata
        Metadata metadata = buildMetadata(index.getId(), basicInfo, 1);
        metadata = metadataDao.saveAndFlush(metadata);

        // 保存MetadataField
        List<MetadataField> metadataFields = createMetadataFields(metadata.getId(), fields);
        metadataFieldDao.save(metadataFields);

        return index.getId();
    }

    /**
     * 更新
     *
     * @param basicInfo
     * @param fields
     * @return
     */
    private long update(IndexDTO basicInfo, List<MetadataFieldDTO> fields) {

        // 获取索引数据
        Index index = Optional.ofNullable(indexDao.findOne(basicInfo.getId()))
                .orElseThrow(() -> new NotFoundException("Index Not Found"));

        // 索引名称不让修改
        if (!index.getIndexName().equals(basicInfo.getIndexName())) {
            throw new FieldException(new FieldExceptionItem("basicInfo.indexName",
                    "index.indexName.can.not.modify"));
        }

        // 引擎不让修改
        if (index.getEngineType() != basicInfo.getEngineType()) {
            throw new FieldException(new FieldExceptionItem("basicInfo.engineType",
                    "index.EngineType.can.not.modify"));
        }

        // 修改允许修改的字段，status、create_by,indexName暂时不做修改
        index.setIntroduction(basicInfo.getIntroduction());
        index.setIndexAliasName(basicInfo.getIndexAliasName());

        // 更新index
        indexDao.saveAndFlush(index);

        // 获取Meta信息
        Metadata metadata = findMetadataByIndexIdAndLastVersion(index.getId());

        // 如果Meta信息和Field信息没有修改，则不处理
        if (isMetadataOrFieldsModified(metadata, basicInfo, fields)) {

            // 如果Meta的状态正常，则需要新建Meta和Field，否则直接更新即可
            if (metadata.getStatus().equals(Status.NORMAL.getValue())) {

                // 保存Metadata
                Metadata newMetadata = buildMetadata(index.getId(), basicInfo,
                        metadata.getConfigVersion() + 1);
                // 修改时host不允许变更，即集群不能变更
                newMetadata.setHost(metadata.getHost());
                newMetadata = metadataDao.saveAndFlush(newMetadata);

                // 保存MetadataField
                List<MetadataField> metadataFields = createMetadataFields(newMetadata.getId(), fields);
                metadataFieldDao.save(metadataFields);
            } else {

                // 修改metadata信息
                metadata.setShardCount(Optional.ofNullable(basicInfo.getShardCount())
                        .orElse(IndexConstants.METADATA_DEFAULT_SHARD_COUNT));
                metadata.setReplicationFactor(Optional.ofNullable(basicInfo.getReplicationFactor())
                        .orElse(IndexConstants.METADATA_DEFAULT_REPLICATION_FACTOR));
                metadata.setMaxShardsPerNode(Optional.ofNullable(basicInfo.getMaxShardsPerNode())
                        .orElse(IndexConstants.METADATA_DEFAULT_MAX_SHARDS_PER_NODE));
                metadata.setConfigName(basicInfo.getIndexName());

                metadataDao.saveAndFlush(metadata);

                // 先删除原来的metadataField，然后再新增
                metadataFieldDao.deleteByMetaId(metadata.getId());
                List<MetadataField> metadataFields = createMetadataFields(metadata.getId(), fields);
                metadataFieldDao.save(metadataFields);
            }
        }

        return index.getId();
    }

    private Index createIndex(IndexDTO basicInfo) {
        Index index = new Index();
        index.setProjectId(basicInfo.getProjectId());
        index.setIndexName(basicInfo.getIndexName());
        index.setIndexAliasName(Optional.ofNullable(basicInfo.getIndexAliasName()).orElse(""));
        index.setEngineType(basicInfo.getEngineType());
        index.setStatus(Status.NORMAL.getValue());
        index.setIntroduction(Optional.ofNullable(basicInfo.getIntroduction()).orElse(""));
        return index;
    }

    private Metadata buildMetadata(long indexId, IndexDTO basicInfo, int configVersion) {
        Metadata metadata = new Metadata();
        metadata.setIndexId(indexId);
        metadata.setStatus(Status.STOP.getValue());
        metadata.setShardCount(Optional.ofNullable(basicInfo.getShardCount())
                .orElse(IndexConstants.METADATA_DEFAULT_SHARD_COUNT));
        metadata.setReplicationFactor(Optional.ofNullable(basicInfo.getReplicationFactor())
                .orElse(IndexConstants.METADATA_DEFAULT_REPLICATION_FACTOR));
        metadata.setMaxShardsPerNode(Optional.ofNullable(basicInfo.getMaxShardsPerNode())
                .orElse(IndexConstants.METADATA_DEFAULT_MAX_SHARDS_PER_NODE));

        EngineType engineType = Enums.findValue(basicInfo.getEngineType(), EngineType.class);
        metadata.setHost(clusterService.getClusterMap(engineType).get(basicInfo.getClusterName()).getHost());
        metadata.setConfigName(basicInfo.getIndexName());
        metadata.setConfigVersion(configVersion);

        // 剩下的字段先设定默认值
        metadata.setAutoAddReplicas(IndexConstants.METADATA_DEFAULT_AUTO_ADD_REPLICAS);
        metadata.setRouterName(IndexConstants.METADATA_DEFAULT_ROUTER_NAME);

        return metadata;
    }

    private List<MetadataField> createMetadataFields(long metadataId, List<MetadataFieldDTO> fields) {
        return fields.stream()
                .map(field -> {
                    MetadataField metadataField = new MetadataField();
                    metadataField.setMetaId(metadataId);
                    metadataField.setFieldName(field.getFieldName());
                    metadataField.setFieldTypeId(field.getFieldTypeId());
                    metadataField.setFieldComment(Optional.of(field.getFieldComment()).orElse(""));
                    metadataField.setUniqueKey(field.getUniqueKey());
                    //加入数据版本号字段
                    metadataField.setVersionKey(field.getVersionKey());
                    metadataField.setRequired(field.getRequired());
                    metadataField.setIndexed(field.getIndexed());
                    metadataField.setStored(field.getStored());
                    metadataField.setTermVectors(field.getTermVectors());
                    metadataField.setMultiValued(field.getMultiValued());
                    // docValues默认为false，以后扩展
                    metadataField.setDocValues(false);
                    metadataField.setDefaultValue(Optional.ofNullable(field.getDefaultValue()).orElse(""));
                    metadataField.setCopyToField(Optional.ofNullable(field.getCopyToField()).orElse(""));
                    metadataField.setBoost(Optional.ofNullable(field.getBoost())
                            .orElse(IndexConstants.METADATA_FIELD_DEFAULT_BOOST));
                    return metadataField;
                })
                .collect(Collectors.toList());
    }

    private Metadata findMetadataByIndexIdAndLastVersion(long indexId) {
        // 获取Meta信息
        List<Metadata> metas = metadataDao.findFirst1ByIndexIdOrderByConfigVersionDesc(indexId);
        if (CollectionUtils.isEmpty(metas)) {
            throw new NotFoundException("index meta not found");
        }
        return metas.get(0);
    }

    private Metadata findMetadataByIndexId(long indexId) {
        // 获取Meta信息
        List<Collection> collections = collectionDao.findByIndexIdInAndStatus(Lists.newArrayList(indexId),
                Status.NORMAL.getValue());
        if (CollectionUtils.isEmpty(collections)) {
            throw new NotFoundException("index collection not found");
        }

        Collection collection = collections.get(0);
        long metaId = collection.getMetaId();
        return Optional.ofNullable(metadataDao.findOne(metaId))
                .orElseThrow(() -> new NotFoundException("index metadata not found"));

    }

    private Metadata findMetadataByIndexIdAndCollectionName(long indexId, String collectionName) {

        List<Collection> collections = collectionDao.findByIndexIdAndCollectionName(indexId, collectionName);
        if (CollectionUtils.isEmpty(collections)) {
            throw new NotFoundException(String.format("index collection %s not found, indexId: %d",
                    collectionName, indexId));
        }

        Collection collection = collections.get(0);
        long metaId = collection.getMetaId();
        return Optional.ofNullable(metadataDao.findOne(metaId))
                .orElseThrow(() -> new NotFoundException(
                        String.format("index metadata not found, indexId: %d, collectionName: %s",
                                indexId, collectionName)));

    }

    private Metadata findMetadataByIndexIdAndStatusNormal(long indexId) {
        // 获取Meta信息
        List<Metadata> metas = metadataDao.findByStatusAndIndexIdIn(Status.NORMAL.getValue(),
                Lists.newArrayList(indexId));
        if (CollectionUtils.isEmpty(metas)) {
            throw new NotFoundException("index meta not found");
        }
        return metas.get(0);
    }

    private boolean isMetadataOrFieldsModified(Metadata metaData, IndexDTO basicInfo, List<MetadataFieldDTO> fields) {

        // meta信息判断
        if (!basicInfo.getMaxShardsPerNode().equals(metaData.getMaxShardsPerNode())
                || !basicInfo.getReplicationFactor().equals(metaData.getReplicationFactor())
                || !basicInfo.getShardCount().equals(metaData.getShardCount())
                || !basicInfo.getIndexName().equals(metaData.getConfigName())) {
            return true;
        }

        // fields判断
        List<MetadataField> metadataFields = Optional.ofNullable(metadataFieldDao.findByMetaId(metaData.getId()))
                .orElse(Collections.emptyList());

        // fields个数不一样则表示修改过
        if (metadataFields.size() != fields.size()) {
            return true;
        }

        Map<String, MetadataField> fieldNameMap = metadataFields.stream()
                .collect(Collectors.toMap(MetadataField::getFieldName, f -> f));
        for (MetadataFieldDTO field : fields) {
            String fieldName = field.getFieldName();
            if (!fieldNameMap.containsKey(fieldName)) {
                return true;
            }
            MetadataField oldField = fieldNameMap.get(fieldName);
            if (!field.getFieldName().equals(oldField.getFieldName())
                    || !field.getFieldTypeId().equals(oldField.getFieldTypeId())
                    || !field.getFieldComment().equals(oldField.getFieldComment())
                    || !field.getUniqueKey().equals(oldField.getUniqueKey())
                    || !field.getVersionKey().equals(oldField.getVersionKey())
                    || !field.getRequired().equals(oldField.getRequired())
                    || !field.getIndexed().equals(oldField.getIndexed())
                    || !field.getStored().equals(oldField.getStored())
                    || !field.getTermVectors().equals(oldField.getTermVectors())
                    || !field.getMultiValued().equals(oldField.getMultiValued())
                    || !field.getDefaultValue().equals(oldField.getDefaultValue())
                    || !field.getCopyToField().equals(oldField.getCopyToField())
                    || !field.getBoost().equals(oldField.getBoost())) {
                return true;
            }
        }

        return false;
    }

    private void fillIndexMetadata(IndexQueryResult data, Metadata metadata) {
        data.setShardCount(metadata.getShardCount());
        data.setMaxShardsPerNode(metadata.getMaxShardsPerNode());
        data.setReplicationFactor(metadata.getReplicationFactor());
        data.setMetadataStatus(metadata.getStatus());
        data.setHost(metadata.getHost());
        data.setConfigVersion(metadata.getConfigVersion());
    }

    private IndexInfoQueryResult createIndexQueryResult(Index index, Metadata metadata, List<MetadataField> fields,
    		SearchUserDto searchUser) {
        IndexInfoQueryResult result = new IndexInfoQueryResult();
        IndexQueryResult basicInfo = BeanCopy.of(index, new IndexQueryResult())
                .copy(BeanUtils::copyProperties)
                .copy((s, t) -> t.setShardCount(metadata.getShardCount()))
                .copy((s, t) -> t.setReplicationFactor(metadata.getReplicationFactor()))
                .copy((s, t) -> t.setMaxShardsPerNode(metadata.getMaxShardsPerNode()))
                .copy((s, t) -> t.setHost(metadata.getHost()))
                .copy((s, t) -> t.setConfigVersion(metadata.getConfigVersion()))
                .copy((s, t) -> {
                    Map<String, ClusterProperties.ClusterInfo> clusterInfoMap =
                            clusterService.getHostKeyClusterMap(Enums.findValue(s.getEngineType(), EngineType.class));
                    t.setClusterName(clusterInfoMap.containsKey(metadata.getHost()) ?
                            clusterInfoMap.get(metadata.getHost()).getName() : "");
                })
                .get();
        basicInfo.setCreateByName(Optional.ofNullable(searchUser)
                .map(SearchUserDto::getUserName)
                .orElse(""));

        result.setBasicInfo(basicInfo);
        result.setFields(fields.stream()
                .map(BeanCopy.toFunction(MetadataFieldQueryResult.class, BeanUtils::copyProperties))
                .collect(Collectors.toList()));
        return result;
    }

    private IndexSchema createIndexSchema(Index index, Metadata metadata, List<MetadataField> fields) {

        IndexSchema indexSchema = new IndexSchema();
        indexSchema.setIndexName(index.getIndexName());

        List<SchemaField> schemaFields = new ArrayList<>(fields.size());
        indexSchema.setFields(schemaFields);

        List<SchemaCopyField> schemaCopyFields = new ArrayList<>(5);
        indexSchema.setCopyFields(schemaCopyFields);

        for (MetadataField field : fields) {

            SchemaField schemaField = new SchemaField();
            schemaField.setIndexed(field.getIndexed());
            schemaField.setMultiValued(field.getMultiValued());
            schemaField.setDocValues(field.getDocValues());
            schemaField.setName(field.getFieldName());
            schemaField.setRemarks(field.getFieldComment());
            schemaField.setRequired(field.getRequired());
            schemaField.setStored(field.getStored());
            schemaField.setType(Enums.find(field.getFieldTypeId(), MetadataFieldType.class)
                    .map(MetadataFieldType::getFieldType)
                    .orElseThrow(() -> new NotFoundException("Index meta field type not found")));
            schemaField.setUniqueKey(field.getUniqueKey());

            // 不是UniqueKey或者不是Required时才能设默认值
            if (!field.getUniqueKey() && !field.getRequired()) {
                schemaField.setDefaultValue(field.getDefaultValue());
            }

            schemaFields.add(schemaField);

            // 设定主键
            if (field.getUniqueKey()) {
                indexSchema.setUniqueKey(schemaField);
            }

            // 设定Copy
            if (StringUtils.isNotEmpty(field.getCopyToField())) {
                String[] copys = field.getCopyToField().split(",");
                for (String copy : copys) {
                    if (StringUtils.isNotEmpty(copy)) {
                        SchemaCopyField copyField = new SchemaCopyField();
                        copyField.setSource(field.getFieldName());
                        copyField.setDest(copy);
                        schemaCopyFields.add(copyField);
                    }
                }
            }
        }

        return indexSchema;
    }

    private String formatConfigNameInZk(Metadata metadata) {
        return String.format("%s_%d", metadata.getConfigName(), metadata.getConfigVersion());
    }

    private String formatCollectionName(Metadata metadata) {
        return String.format("%s_%d_%s", metadata.getConfigName(), metadata.getConfigVersion(),
                DateFormatUtils.format(new Date(), "yyyyMMddHH"));
    }

    private IndexMeta indexInfo2IndexMeta(Index index, Metadata metadata, List<MetadataField> metadataFields) {
        IndexMeta indexMeta = new IndexMeta();
        indexMeta.setIndexName(index.getIndexName());
        indexMeta.setHost(metadata.getHost());
        indexMeta.setEngineType(index.getEngineType());
        indexMeta.setShardCount(metadata.getShardCount());
        indexMeta.setReplicationFactor(metadata.getReplicationFactor());

        Map<String, FieldMeta> filedMetas = new HashMap<String, FieldMeta>(metadataFields.size() * 2);
        for (MetadataField metadataField : metadataFields) {
            FieldMeta fieldMeta = new FieldMeta();
            String name = metadataField.getFieldName();
            fieldMeta.setName(name);
            MetadataFieldType metadataFieldType = Enums.find(metadataField.getFieldTypeId(), MetadataFieldType.class)
                    .orElse(MetadataFieldType.STRING);
            fieldMeta.setType(metadataFieldType);
            fieldMeta.setPk(metadataField.getUniqueKey());
            fieldMeta.setMultiValued(metadataField.getMultiValued());
            fieldMeta.setRequired(metadataField.getRequired());
            //设置是否被索引
            fieldMeta.setIndexed(metadataField.getIndexed());
            if (metadataField.getUniqueKey()) {
                indexMeta.setPk(fieldMeta);
            }

            filedMetas.put(name, fieldMeta);
        }
        indexMeta.setFiledMetas(filedMetas);

        return indexMeta;
    }
}
